package net.bwie.realtime.jtp.dws.douyin.log.functions;

import net.bwie.realtime.jtp.dws.douyin.log.bean.DealUserRegionMetric4;
import net.bwie.realtime.jtp.dws.douyin.log.bean.EventLog6;
import net.bwie.realtime.jtp.dws.douyin.log.bean.DealUserRegionMetric4;
import net.bwie.realtime.jtp.dws.douyin.log.bean.EventLog6;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;

import static java.awt.SystemColor.window;

public class DouYinUserRegionFunction4 implements
        WindowFunction<EventLog6, DealUserRegionMetric4, Long, TimeWindow> {

    // 日期格式化工具（线程安全处理）
    private static final ThreadLocal<SimpleDateFormat> DATE_FORMATTER = ThreadLocal.withInitial(
            () -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    );
    private static final ThreadLocal<SimpleDateFormat> DATE_ONLY_FORMATTER = ThreadLocal.withInitial(
            () -> new SimpleDateFormat("yyyy-MM-dd")
    );

    @Override
    public void apply(
            Long anchorId,  // 分组键：主播ID
            TimeWindow window,
            Iterable<EventLog6> input,
            Collector<DealUserRegionMetric4> out) {

        // 1. 初始化统计容器
        // key: 地区名称 value: 该地区的订单统计（第一个元素：订单数，第二个元素：去重用户ID）
        Map<String, RegionStat> regionStatMap = new HashMap<>();

        // 2. 遍历窗口内所有事件，聚合统计数据
        for (EventLog6 event : input) {
            String region = event.getRegion();
            Long userId = event.getUserId();
            Long liveRoomId = event.getLiveRoomId();

            // 初始化地区统计对象
            RegionStat stat = regionStatMap.computeIfAbsent(region, k -> new RegionStat(liveRoomId));

            // 累加订单数
            stat.dealOrderCount++;

            // 记录去重用户ID
            stat.userIds.add(userId);
        }

        // 3. 计算全量统计（总订单数、总用户数）
        int totalDealOrderCount = regionStatMap.values().stream()
                .mapToInt(stat -> stat.dealOrderCount)
                .sum();

        long totalDealUserUv = regionStatMap.values().stream()
                .flatMap(stat -> stat.userIds.stream())
                .distinct()
                .count();

        // 4. 对地区按订单数排序（降序），用于计算排名
        List<Map.Entry<String, RegionStat>> sortedRegions = regionStatMap.entrySet().stream()
                .sorted((e1, e2) -> Integer.compare(e2.getValue().dealOrderCount, e1.getValue().dealOrderCount))
                .collect(Collectors.toList());

        // 5. 格式化窗口时间
        String windowStartTime = DATE_FORMATTER.get().format(new Date(window.getStart()));
        String windowEndTime = DATE_FORMATTER.get().format(new Date(window.getEnd()));
        String curDate = DATE_ONLY_FORMATTER.get().format(new Date(window.getStart()));

        // 6. 生成各地区的指标数据并输出（仅保留TOP5）
        for (int i = 0; i < sortedRegions.size(); i++) {
            Map.Entry<String, RegionStat> entry = sortedRegions.get(i);
            String region = entry.getKey();
            RegionStat stat = entry.getValue();

            // 计算地区占比（避免除零异常）
            double regionRatio = totalDealOrderCount > 0
                    ? (double) stat.dealOrderCount / totalDealOrderCount
                    : 0.0;

            // 构建指标对象
            DealUserRegionMetric4 metric = new DealUserRegionMetric4();
            metric.setWindowStartTime(windowStartTime);
            metric.setWindowEndTime(windowEndTime);
            metric.setCurDate(curDate);
            metric.setAnchorId(anchorId);
            metric.setLiveRoomId(stat.liveRoomId);
            metric.setRegion(region);
            metric.setDealUserUv((long) stat.userIds.size());
            metric.setTotalDealUserUv(totalDealUserUv);
            metric.setRegionRatio(regionRatio);
            metric.setRegionRank(i + 1);  // 排名从1开始

            // 仅输出排名1-5的数据（TOP5）
            if (metric.getRegionRank() >= 1 && metric.getRegionRank() <= 5) {
                out.collect(metric);
            }
        }
    }

    /**
     * 地区统计辅助类
     * 存储单个地区的订单数、去重用户ID和直播间ID
     */
    private static class RegionStat {
        int dealOrderCount;        // 订单数
        Set<Long> userIds;         // 去重用户ID集合
        Long liveRoomId;           // 直播间ID

        public RegionStat(Long liveRoomId) {
            this.dealOrderCount = 0;
            this.userIds = new HashSet<>();
            this.liveRoomId = liveRoomId;
        }
    }
}
